package com.mashibing.rocketmq;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;

import java.util.List;

/**
 * @Author: JLL
 * @Date: 2020/2/29 20:15
 */
public class Consumer {

    public static void main(String[] args) throws Exception{
        //这好比是一个group
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("xxoocsm");

        consumer.setNamesrvAddr("192.168.150.133:9876");

        // 每个consumer 只能关注（或者叫订阅 subscribe）一个topic
        /**
         * 参数一： topic 关注的消息的地址
         *  参数二： 过滤器（必须要有）     * 表示不过滤
         */
        consumer.subscribe("myTopic001", "*");

        //注册一个监听器，监听到了就走回调函数
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                for (MessageExt msg : msgs) {
                    System.out.println(new String(msg.getBody()));
                }
                // 默认情况下 这条消息只会被 一个consumer 消费到 点对点
                // message 状态修改、你告诉broker我接收到了消息，broker就不会把消息再退给别人了
                // 其实这就是 ack
                /**
                 * return的是状态信息。
                 * CONSUME_SUCCESS 表示收到了消息，然后跟服务器说一声撒。
                 */
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        consumer.start();

        System.out.println("Consumer 02 start...");
    }
}
